package com.ccut.xunfeiSpringboot.service.impl;

import com.alibaba.fastjson2.JSON;
import com.ccut.xunfeiSpringboot.api.SparkChatClient;
import com.ccut.xunfeiSpringboot.common.*;
import com.ccut.xunfeiSpringboot.constant.SparkModel;
import com.ccut.xunfeiSpringboot.dto.*;
import com.ccut.xunfeiSpringboot.dto.response.SparkChatResponse;
import com.ccut.xunfeiSpringboot.entity.CalendarEvent;
import com.ccut.xunfeiSpringboot.entity.PositionInformation;
import com.ccut.xunfeiSpringboot.entity.RecommendCareers;
import com.ccut.xunfeiSpringboot.mapper.BaseMapper;
import com.ccut.xunfeiSpringboot.mapper.CalendarEventMapper;
import com.ccut.xunfeiSpringboot.mapper.JobMapper;
import com.ccut.xunfeiSpringboot.mapper.RecommendedJobsMapper;
import com.ccut.xunfeiSpringboot.model.sparkmodel.SparkChatParam;
import com.ccut.xunfeiSpringboot.model.RoleContent;
import com.ccut.xunfeiSpringboot.model.sparkmodel.WebSearch;
import com.ccut.xunfeiSpringboot.service.AbstractSparkModelWebSocketListener;
import com.ccut.xunfeiSpringboot.service.BaseService;
import lombok.RequiredArgsConstructor;
import okhttp3.Response;
import okhttp3.WebSocket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.net.MalformedURLException;
import java.security.SignatureException;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

@Service
@RequiredArgsConstructor
public class BaseServiceImpl implements BaseService {

    @Autowired
    private BaseMapper baseMapper;

    @Autowired
    private RecommendedJobsMapper recommendedJobsMapper;

    @Autowired
    private JobMapper jobMapper;

    @Autowired
    private CalendarEventMapper calendarEventMapper;

    private final XunfeiBigModelConfig config;
    private final Logger logger = LoggerFactory.getLogger(BaseServiceImpl.class);

    // 新增字段用于流式输出
    private Consumer<String> streamConsumer;
    // 用于拼接不完整的 JSON 内容
    private StringBuilder partialContent = new StringBuilder();

    /**
     * 提取公共的 SparkChatClient 构建逻辑
     */
    private SparkChatClient buildSparkChatClient(ChatRequest request) {
        WebSearch webSearch = new WebSearch();
        webSearch.setEnable(true);
        if (request.getDeepThinking()) {
            webSearch.setSearchMode("deep");
        }
        if (request.getNetworkSearch()) {
            webSearch.setShowRefLabel(true);
        }

        return new SparkChatClient.Builder()
                .signatureWs(config.getAppid(), config.getApiKey(), config.getApiSecret(), SparkModel.SPARK_4_0_ULTRA)
                .webSearch(webSearch)
                .topK(6)
                .temperature(1.0F)
                .callTimeout(3, TimeUnit.MINUTES)
                .build();
    }

    /**
     * 提取公共的消息构建逻辑
     */
    private SparkChatParam buildSparkChatParam(ChatRequest request) {
        List<RoleContent> messages = request.getMessages();

        return SparkChatParam.builder()
                .messages(messages)
                .chatId("test_001")
                .build();
    }

    /**
     * 流式发送职位建议信息
     */
    @Override
    public void streamJobSuggestions(ChatRequest request, Consumer<String> contentConsumer) throws Exception {
        this.streamConsumer = contentConsumer;

        SparkChatClient client = buildSparkChatClient(request);
        SparkChatParam sendParam = buildSparkChatParam(request);

        CountDownLatch latch = new CountDownLatch(1);

        client.send(sendParam, new AbstractSparkModelWebSocketListener() {

            @Override
            public void onSuccess(WebSocket webSocket, SparkChatResponse resp) {
                if (resp.getHeader().getCode() != 0) {
                    logger.error("错误码: {}, 错误信息: {}", resp.getHeader().getCode(), resp.getHeader().getMessage());
                    webSocket.close(1000, "错误码终止");
                    latch.countDown();
                    return;
                }

                if (resp.getPayload() != null && resp.getPayload().getChoices() != null) {
                    List<SparkChatResponse.Payload.Choices.Text> texts = resp.getPayload().getChoices().getText();
                    if (texts != null && !texts.isEmpty()) {
                        for (SparkChatResponse.Payload.Choices.Text text : texts) {
                            String content = text.getContent();
                            logger.info("[WebSocket]接收到内容: {}", content);
                            partialContent.append(content);

                            // 尝试提取并消费完整的职位建议信息
                            processJobSuggestionContent(partialContent, streamConsumer,request.getUserId());
                        }
                    }

                    // 当状态为 2 表示响应结束，此时关闭 WebSocket 并释放 latch
                    if (resp.getPayload().getChoices().getStatus() == 2) {
                        // 最后再尝试处理一次缓冲区内容
                        processJobSuggestionContent(partialContent, streamConsumer,request.getUserId());

                        logger.info("[WebSocket]会话结束，已发送所有职位建议");
                        webSocket.close(1000, "正常关闭");
                        latch.countDown();
                    }
                }
            }

            @Override
            public void onFail(WebSocket webSocket, Throwable t, Response response) {
                logger.error("WebSocket 请求失败: ", t);
                webSocket.close(1000, t.getMessage());
                latch.countDown();
            }

            @Override
            public void onClose(WebSocket webSocket, int code, String reason) {
                logger.info("WebSocket 连接已关闭，原因: {}", reason);
            }
        });

        boolean completed = latch.await(3, TimeUnit.MINUTES);
        if (!completed) {
            throw new RuntimeException("WebSocket 请求超时");
        }
    }

    @Override
    public void arrangeEvents(ChatRequest chatRequest, Consumer<String> contentConsumer) throws InterruptedException, MalformedURLException, SignatureException {
        this.streamConsumer = contentConsumer;

        SparkChatClient client = buildSparkChatClient(chatRequest);
        SparkChatParam sendParam = buildSparkChatParam(chatRequest);
        CountDownLatch latch = new CountDownLatch(1);
        client.send(sendParam, new AbstractSparkModelWebSocketListener() {
            @Override
            public void onSuccess(WebSocket webSocket, SparkChatResponse resp) {
                if (resp.getHeader().getCode() != 0) {
                    logger.error("错误码: {}, 错误信息: {}", resp.getHeader().getCode(), resp.getHeader().getMessage());
                    webSocket.close(1000, "错误码终止");
                    latch.countDown();
                    return;
                }
                if (resp.getPayload() != null && resp.getPayload().getChoices() != null) {
                    List<SparkChatResponse.Payload.Choices.Text> texts = resp.getPayload().getChoices().getText();
                    if (texts != null && !texts.isEmpty()) {
                        for (SparkChatResponse.Payload.Choices.Text text : texts) {
                            String content = text.getContent();
                            logger.info("[WebSocket]接收到内容: {}", content);
                            partialContent.append(content);
                            // 尝试提取并消费完整的日历事件信息
                            processEventContent(partialContent, streamConsumer, chatRequest.getUserId());
                        }
                    }
                    // 当状态为 2 表示响应结束，此时关闭 WebSocket 并释放 latch
                    if (resp.getPayload().getChoices().getStatus() == 2) {
                        // 最后再尝试处理一次缓冲区内容
                        processEventContent(partialContent, streamConsumer, chatRequest.getUserId());
                        logger.info("[WebSocket]会话结束，已发送所有日历事件");
                        webSocket.close(1000, "正常关闭");
                        latch.countDown();
                    }
                }
            }

            @Override
            public void onFail(WebSocket webSocket, Throwable t, Response response) {
                logger.error("WebSocket 请求失败: ", t);
                webSocket.close(1000, t.getMessage());
                latch.countDown();
            }

            @Override
            public void onClose(WebSocket webSocket, int code, String reason) {
                logger.info("WebSocket 连接已关闭，原因: {}", reason);
            }
        });
        boolean completed = latch.await(3, TimeUnit.MINUTES);
        if (!completed) {
            throw new RuntimeException("WebSocket 请求超时");
        }
    }

    @Override
    public KnowledgeGraphDTO recommendKnowledgeGraph(ChatRequest chatRequest) throws MalformedURLException, SignatureException {
        SparkChatClient client = buildSparkChatClient(chatRequest);
        SparkChatParam sendParam = buildSparkChatParam(chatRequest);
        final StringBuilder responseContent = new StringBuilder(); // 用于存储最终返回的内容

        CountDownLatch latch = new CountDownLatch(1);
        client.send(sendParam, new AbstractSparkModelWebSocketListener() {
            @Override
            public void onSuccess(WebSocket webSocket, SparkChatResponse resp) {
                if (resp.getHeader().getCode() != 0) {
                    logger.error("错误码: {}, 错误信息: {}", resp.getHeader().getCode(), resp.getHeader().getMessage());
                    webSocket.close(1000, "错误码终止");
                    latch.countDown();
                    return;
                }
                if (resp.getPayload() != null && resp.getPayload().getChoices() != null) {
                    List<SparkChatResponse.Payload.Choices.Text> texts = resp.getPayload().getChoices().getText();
                    if (texts != null && !texts.isEmpty()) {
                        for (SparkChatResponse.Payload.Choices.Text text : texts) {
                            String content = text.getContent();
                            responseContent.append(content); // 将每次获取到的内容添加到最终结果中
                        }
                    }
                    // 当状态为 2 表示响应结束，此时关闭 WebSocket 并释放 latch
                    if (resp.getPayload().getChoices().getStatus() == 2) {
                        logger.info("[WebSocket]会话结束，已接收所有知识图谱推荐内容");
                        webSocket.close(1000, "正常关闭");
                        latch.countDown();
                    }
                }
            }

            @Override
            public void onFail(WebSocket webSocket, Throwable t, Response response) {
                logger.error("WebSocket 请求失败: ", t);
                webSocket.close(1000, t.getMessage());
                latch.countDown();
            }

            @Override
            public void onClose(WebSocket webSocket, int code, String reason) {
                logger.info("WebSocket 连接已关闭，原因: {}", reason);
            }
        });

        try {
            boolean completed = latch.await(3, TimeUnit.MINUTES);
            if (!completed) {
                throw new RuntimeException("WebSocket 请求超时");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            logger.error("线程被中断", e);
            return null;
        }

        logger.info("[WebSocket]原始返回内容: {}", responseContent);

        // 截取第一个{到最后一个}之间的完整JSON内容
        String responseStr = responseContent.toString();
        int startIndex = responseStr.indexOf('{');
        int endIndex = responseStr.lastIndexOf('}');

        if (startIndex != -1 && endIndex != -1 && endIndex > startIndex) {
            String jsonContent = responseStr.substring(startIndex, endIndex + 1);
            logger.info("[WebSocket]提取的JSON内容: {}", jsonContent);
            return JSON.parseObject(jsonContent, KnowledgeGraphDTO.class);
        } else {
            logger.warn("无法找到完整的JSON对象，原始内容: {}", responseStr);
            return null; // 返回空JSON对象作为默认值
        }
    }

    @Override
    public AnalyseResumeDTO analyseResume(ChatRequest chatRequest) throws MalformedURLException, SignatureException {
        SparkChatClient client = buildSparkChatClient(chatRequest);
        SparkChatParam sendParam = buildSparkChatParam(chatRequest);
        final StringBuilder responseContent = new StringBuilder(); // 用于存储最终返回的内容
        CountDownLatch latch = new CountDownLatch(1);
        client.send(sendParam, new AbstractSparkModelWebSocketListener() {
            @Override
            public void onSuccess(WebSocket webSocket, SparkChatResponse resp) {
                if (resp.getHeader().getCode() != 0) {
                    logger.error("错误码: {}, 错误信息: {}", resp.getHeader().getCode(), resp.getHeader().getMessage());
                    webSocket.close(1000, "错误码终止");
                    latch.countDown();
                    return;
                }
                if (resp.getPayload() != null && resp.getPayload().getChoices() != null) {
                    List<SparkChatResponse.Payload.Choices.Text> texts = resp.getPayload().getChoices().getText();
                    if (texts != null && !texts.isEmpty()) {
                        for (SparkChatResponse.Payload.Choices.Text text : texts) {
                            String content = text.getContent();
                            responseContent.append(content); // 将每次获取到的内容添加到最终结果中
                        }
                    }
                    // 当状态为 2 表示响应结束，此时关闭 WebSocket 并释放 latch
                    if (resp.getPayload().getChoices().getStatus() == 2) {
                        logger.info("[WebSocket]会话结束，已接收所有简历分析内容");
                        webSocket.close(1000, "正常关闭");
                        latch.countDown();
                    }
                }
            }

            @Override
            public void onFail(WebSocket webSocket, Throwable t, Response response) {
                logger.error("WebSocket 请求失败: ", t);
                webSocket.close(1000, t.getMessage());
                latch.countDown();
            }

            @Override
            public void onClose(WebSocket webSocket, int code, String reason) {
                logger.info("WebSocket 连接已关闭，原因: {}", reason);
            }
        });
        try {
            boolean completed = latch.await(3, TimeUnit.MINUTES);
            if (!completed) {
                throw new RuntimeException("WebSocket 请求超时");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            logger.error("线程被中断", e);
            return null;
        }
        logger.info("[WebSocket]原始返回内容: {}", responseContent);
        // 截取第一个{到最后一个}之间的完整JSON内容
        String responseStr = responseContent.toString();
        int startIndex = responseStr.indexOf('{');
        int endIndex = responseStr.lastIndexOf('}');
        if (startIndex != -1 && endIndex != -1 && endIndex > startIndex) {
            String jsonContent = responseStr.substring(startIndex, endIndex + 1);
            logger.info("[WebSocket]提取的JSON内容: {}", jsonContent);
            return JSON.parseObject(jsonContent, AnalyseResumeDTO.class);
        } else {
            logger.warn("无法找到完整的JSON对象，原始内容: {}", responseStr);
            return null; // 返回空JSON对象作为默认值
        }
    }

    @Override
    public AnalyseCapabilityDTO analyseCapability(ChatRequest chatRequest) throws MalformedURLException, SignatureException {
        SparkChatClient client = buildSparkChatClient(chatRequest);
        SparkChatParam sendParam = buildSparkChatParam(chatRequest);
        final StringBuilder responseContent = new StringBuilder(); // 用于存储最终返回的内容
        CountDownLatch latch = new CountDownLatch(1);
        client.send(sendParam, new AbstractSparkModelWebSocketListener() {
            @Override
            public void onSuccess(WebSocket webSocket, SparkChatResponse resp) {
                if (resp.getHeader().getCode() != 0) {
                    logger.error("错误码: {}, 错误信息: {}", resp.getHeader().getCode(), resp.getHeader().getMessage());
                    webSocket.close(1000, "错误码终止");
                    latch.countDown();
                    return;
                }
                if (resp.getPayload() != null && resp.getPayload().getChoices() != null) {
                    List<SparkChatResponse.Payload.Choices.Text> texts = resp.getPayload().getChoices().getText();
                    if (texts != null && !texts.isEmpty()) {
                        for (SparkChatResponse.Payload.Choices.Text text : texts) {
                            String content = text.getContent();
                            responseContent.append(content); // 将每次获取到的内容添加到最终结果中
                        }
                    }
                    // 当状态为 2 表示响应结束，此时关闭 WebSocket 并释放 latch
                    if (resp.getPayload().getChoices().getStatus() == 2) {
                        logger.info("[WebSocket]会话结束，已接收所有能力分析内容");
                        webSocket.close(1000, "正常关闭");
                        latch.countDown();
                    }
                }
            }

            @Override
            public void onFail(WebSocket webSocket, Throwable t, Response response) {
                logger.error("WebSocket 请求失败: ", t);
                webSocket.close(1000, t.getMessage());
                latch.countDown();
            }

            @Override
            public void onClose(WebSocket webSocket, int code, String reason) {
                logger.info("WebSocket 连接已关闭，原因: {}", reason);
            }
        });
        try {
            boolean completed = latch.await(3, TimeUnit.MINUTES);
            if (!completed) {
                throw new RuntimeException("WebSocket 请求超时");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            logger.error("线程被中断", e);
            return null;
        }
        logger.info("[WebSocket]原始返回内容: {}", responseContent);
        // 截取第一个{到最后一个}之间的完整JSON内容
        String responseStr = responseContent.toString();
        int startIndex = responseStr.indexOf('{');
        int endIndex = responseStr.lastIndexOf('}');
        if (startIndex != -1 && endIndex != -1 && endIndex > startIndex) {
            String jsonContent = responseStr.substring(startIndex, endIndex + 1);
            logger.info("[WebSocket]提取的JSON内容: {}", jsonContent);
            return JSON.parseObject(jsonContent, AnalyseCapabilityDTO.class);
        } else {
            logger.warn("无法找到完整的JSON对象，原始内容: {}", responseStr);
            return null; // 返回空JSON对象作为默认值
        }
    }

    @Override
    public UserAnswerAnalyseDTO analyseUserAnswer(ChatRequest chatRequest) throws MalformedURLException, SignatureException {
        SparkChatClient client = buildSparkChatClient(chatRequest);
        SparkChatParam sendParam = buildSparkChatParam(chatRequest);
        final StringBuilder responseContent = new StringBuilder(); // 用于存储最终返回的内容
        CountDownLatch latch = new CountDownLatch(1);
        client.send(sendParam, new AbstractSparkModelWebSocketListener() {
            @Override
            public void onSuccess(WebSocket webSocket, SparkChatResponse resp) {
                if (resp.getHeader().getCode() != 0) {
                    logger.error("错误码: {}, 错误信息: {}", resp.getHeader().getCode(), resp.getHeader().getMessage());
                    webSocket.close(1000, "错误码终止");
                    latch.countDown();
                    return;
                }
                if (resp.getPayload() != null && resp.getPayload().getChoices() != null) {
                    List<SparkChatResponse.Payload.Choices.Text> texts = resp.getPayload().getChoices().getText();
                    if (texts != null && !texts.isEmpty()) {
                        for (SparkChatResponse.Payload.Choices.Text text : texts) {
                            String content = text.getContent();
                            responseContent.append(content); // 将每次获取到的内容添加到最终结果中
                        }
                    }
                    // 当状态为 2 表示响应结束，此时关闭 WebSocket 并释放 latch
                    if (resp.getPayload().getChoices().getStatus() == 2) {
                        logger.info("[WebSocket]会话结束，已接收所有用户回答分析内容");
                        webSocket.close(1000, "正常关闭");
                        latch.countDown();
                    }
                }
            }

            @Override
            public void onFail(WebSocket webSocket, Throwable t, Response response) {
                logger.error("WebSocket 请求失败: ", t);
                webSocket.close(1000, t.getMessage());
                latch.countDown();
            }

            @Override
            public void onClose(WebSocket webSocket, int code, String reason) {
                logger.info("WebSocket 连接已关闭，原因: {}", reason);
            }
        });
        try {
            boolean completed = latch.await(3, TimeUnit.MINUTES);
            if (!completed) {
                throw new RuntimeException("WebSocket 请求超时");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            logger.error("线程被中断", e);
            return null;
        }
        logger.info("[WebSocket]原始返回内容: {}", responseContent);
        // 截取第一个{到最后一个}之间的完整JSON内容
        String responseStr = responseContent.toString();
        int startIndex = responseStr.indexOf('{');
        int endIndex = responseStr.lastIndexOf('}');
        if (startIndex != -1 && endIndex != -1 && endIndex > startIndex) {
            String jsonContent = responseStr.substring(startIndex, endIndex + 1);
            logger.info("[WebSocket]提取的JSON内容: {}", jsonContent);
            return JSON.parseObject(jsonContent, UserAnswerAnalyseDTO.class);
        } else {
            logger.warn("无法找到完整的JSON对象，原始内容: {}", responseStr);
            return null; // 返回空JSON对象作为默认值
        }
    }

    /**
     * 解析缓冲区内容，提取符合 CalendarEvents 结构的完整 JSON 对象
     */
    private void processEventContent(StringBuilder buffer, Consumer<String> consumer, Long userId) {
        String content = buffer.toString();
        int openBraceIndex = content.indexOf('{');
        while (openBraceIndex != -1) {
            int closeBraceIndex = findClosingBraceMatch(content, openBraceIndex);
            if (closeBraceIndex == -1) {
                break; // 没有找到匹配的闭合括号，等待更多数据
            }
            String possibleJson = content.substring(openBraceIndex, closeBraceIndex + 1);
            try {
                // 假设 CalendarEvents 类已经定义好
                JSON.parseObject(possibleJson, CalendarEvent.class);
                CalendarEvent event = JSON.parseObject(possibleJson, CalendarEvent.class);
                event.setId(System.currentTimeMillis());
                event.setUserId(Math.toIntExact(userId));
                if (event.getRrule() == null){
                    event.setRrule("");
                }
                calendarEventMapper.insert(event);
                String jsonString = JSON.toJSONString(event);
                consumer.accept(jsonString); // 发送给前端
                logger.info("实时发送日历事件 ==> {}", jsonString);
                // 更新缓冲区为剩余部分
                buffer.setLength(0);
                buffer.append(content.substring(closeBraceIndex + 1));
                content = buffer.toString(); // 更新 content 以便继续循环
                openBraceIndex = content.indexOf('{'); // 重置起始索引
            } catch (Exception e) {
                // 如果不符合 CalendarEvents 结构，跳过该段落并继续查找下一个对象
                openBraceIndex = content.indexOf('{', openBraceIndex + 1);
            }
        }
    }

    /**
     * 解析缓冲区内容，提取符合 JobSuggestion 结构的完整 JSON 对象
     */
    private void processJobSuggestionContent(StringBuilder buffer, Consumer<String> consumer,Long userId) {
        String content = buffer.toString();

        int openBraceIndex = content.indexOf('{');
        while (openBraceIndex != -1) {
            int closeBraceIndex = findClosingBraceMatch(content, openBraceIndex);
            if (closeBraceIndex == -1) {
                break; // 没有找到匹配的闭合括号，等待更多数据
            }

            String possibleJson = content.substring(openBraceIndex, closeBraceIndex + 1);
            try {
                // 验证是否符合 JobSuggestion 结构
                JSON.parseObject(possibleJson, RecommendCareers.class);
                RecommendCareers job = JSON.parseObject(possibleJson, RecommendCareers.class);
                job.setUserId(userId);
                consumer.accept(possibleJson); // 发送给前端
                logger.info("实时发送职位建议 ==> {}", possibleJson);
//                if (count++==0){
//                    recommendedJobsMapper.updateRecommendedJobs(job);
//                }else {
                recommendedJobsMapper.insertRecommendedJobs(job);
//                }

                // 更新缓冲区为剩余部分
                buffer.setLength(0);
                buffer.append(content.substring(closeBraceIndex + 1));
                content = buffer.toString(); // 更新 content 以便继续循环
                openBraceIndex = content.indexOf('{'); // 重置起始索引
            } catch (Exception e) {
                // 如果不符合 JobSuggestion 结构，跳过该段落并继续查找下一个对象
                openBraceIndex = content.indexOf('{', openBraceIndex + 1);
            }
        }
    }

    /**
     * 支持流式返回岗位数据的方法
     */
    @Override
    public void searchJobs(ChatRequest request, Consumer<String> contentConsumer) throws Exception {
        this.streamConsumer = contentConsumer; // 设置流式输出

        SparkChatClient client = buildSparkChatClient(request);
        SparkChatParam sendParam = buildSparkChatParam(request);

        CountDownLatch latch = new CountDownLatch(1);

        client.send(sendParam, new AbstractSparkModelWebSocketListener() {

            @Override
            public void onSuccess(WebSocket webSocket, SparkChatResponse resp) {
                if (resp.getHeader().getCode() != 0) {
                    logger.error("错误码: {}, 错误信息: {}", resp.getHeader().getCode(), resp.getHeader().getMessage());
                    webSocket.close(1000, "错误码终止");
                    latch.countDown();
                    return;
                }

                if (resp.getPayload() != null && resp.getPayload().getChoices() != null) {
                    List<SparkChatResponse.Payload.Choices.Text> texts = resp.getPayload().getChoices().getText();
                    if (texts != null && !texts.isEmpty()) {
                        for (SparkChatResponse.Payload.Choices.Text text : texts) {
                            String content = text.getContent();
                            logger.info("[WebSocket]接收到内容: {}", content);
                            partialContent.append(content);

                            // 尝试提取并消费完整的岗位信息
                            processPartialContent(partialContent, streamConsumer, request.getUserId());
                        }
                    }

                    // 当状态为 2 表示响应结束，此时关闭 WebSocket 并释放 latch
                    if (resp.getPayload().getChoices().getStatus() == 2) {
                        // 最后再尝试处理一次缓冲区内容
                        processPartialContent(partialContent, streamConsumer, request.getUserId());

                        logger.info("[WebSocket]会话结束，已发送所有岗位信息");
                        webSocket.close(1000, "正常关闭");
                        latch.countDown();
                    }
                }
            }

            @Override
            public void onFail(WebSocket webSocket, Throwable t, Response response) {
                logger.error("WebSocket 请求失败: ", t);
                webSocket.close(1000, t.getMessage());
                latch.countDown();
            }

            @Override
            public void onClose(WebSocket webSocket, int code, String reason) {
                logger.info("WebSocket 连接已关闭，原因: {}", reason);
            }
        });

        boolean completed = latch.await(3, TimeUnit.MINUTES);
        if (!completed) {
            throw new RuntimeException("WebSocket 请求超时");
        }
    }

    private void processPartialContent(StringBuilder buffer, Consumer<String> consumer, Long userId) {
        String content = buffer.toString();

        int openBraceIndex = content.indexOf('{');
        while (openBraceIndex != -1) {
            int closeBraceIndex = findClosingBraceMatch(content, openBraceIndex);
            if (closeBraceIndex == -1) {
                break; // 没有找到匹配的闭合括号，等待更多数据
            }

            String possibleJson = content.substring(openBraceIndex, closeBraceIndex + 1);
            try {
                // 验证是否符合 PositionInformation 结构
                JSON.parseObject(possibleJson, PositionInformation.class);
                PositionInformation positionInformation = JSON.parseObject(possibleJson, PositionInformation.class);
                positionInformation.setUserId(userId);
                positionInformation.setNumberOfInterviews(0L);
                jobMapper.addJob(positionInformation);
                consumer.accept(possibleJson); // 发送给前端
                logger.info("实时发送岗位信息 ==> {}", possibleJson);

                // 更新缓冲区为剩余部分
                buffer.setLength(0);
                buffer.append(content.substring(closeBraceIndex + 1));
                content = buffer.toString(); // 更新 content 以便继续循环
                openBraceIndex = content.indexOf('{'); // 重置起始索引
            } catch (Exception e) {
//                logger.warn("JSON 解析失败或不符合岗位结构，继续等待更多数据...");
                break;
            }
        }
    }

    /**
     * 查找匹配的右括号位置
     */
    private int findClosingBraceMatch(String content, int start) {
        int depth = 0;
        for (int i = start; i < content.length(); i++) {
            char c = content.charAt(i);
            if (c == '{') {
                depth++;
            } else if (c == '}') {
                depth--;
                if (depth == 0) {
                    return i;
                }
            }
        }
        return -1; // 没有匹配的右括号
    }

    @Override
    public void chatWithStream(ChatRequest request, Consumer<String> contentConsumer) throws Exception {
        SparkChatClient client = buildSparkChatClient(request);
        SparkChatParam sendParam = buildSparkChatParamForChat(request);

        CountDownLatch latch = new CountDownLatch(1);

        client.send(sendParam, new AbstractSparkModelWebSocketListener() {

            @Override
            public void onSuccess(WebSocket webSocket, SparkChatResponse resp) {
                if (resp.getHeader().getCode() != 0) {
                    logger.error("错误码: {}, 错误信息: {}", resp.getHeader().getCode(), resp.getHeader().getMessage());
                    latch.countDown();
                    return;
                }

                if (resp.getPayload() == null && resp.getPayload().getPlugins() == null){
                    List<SparkChatResponse.Payload.Plugin.Text> pluginTexts = resp.getPayload().getPlugins().getText();
                    if (pluginTexts != null && !pluginTexts.isEmpty()){
                        for (SparkChatResponse.Payload.Plugin.Text text : pluginTexts) {
                            String content = text.getContent();
                            if (content != null && !content.isEmpty()) {
                                contentConsumer.accept(content);
                                logger.info("发送内容 ==> {}", content);
                            }
                        }
                    }
                }

                if (resp.getPayload() != null && resp.getPayload().getChoices() != null) {
                    List<SparkChatResponse.Payload.Choices.Text> texts = resp.getPayload().getChoices().getText();
                    if (texts != null && !texts.isEmpty()) {
                        for (SparkChatResponse.Payload.Choices.Text text : texts) {
                            String content = text.getContent();
                            if (content != null && !content.isEmpty()) {
                                contentConsumer.accept(content);
                                logger.info("发送内容 ==> {}", content);
                            }
                        }
                    }


                    if (resp.getPayload().getChoices().getStatus() == 2) {
                        logger.info("[WebSocket]会话结束");
                        webSocket.close(1000, "正常关闭");
                        latch.countDown();
                    }
                }
            }

            @Override
            public void onFail(WebSocket webSocket, Throwable t, Response response) {
                logger.error("WebSocket 请求失败: ", t);
                webSocket.close(1000, t.getMessage());
                latch.countDown();
            }

            @Override
            public void onClose(WebSocket webSocket, int code, String reason) {
                logger.info("WebSocket 连接已关闭，原因: {}", reason);
            }
        });

        boolean completed = latch.await(3, TimeUnit.MINUTES);
        if (!completed) {
            throw new RuntimeException("WebSocket 请求超时");
        }
    }

    /**
     * 专用于 chatWithStream 的消息构建逻辑
     */
    private SparkChatParam buildSparkChatParamForChat(ChatRequest request) {
        List<RoleContent> messages = request.getMessages();
        RoleContent roleContent = messages.get(0);
        roleContent.setContent("你是一个聊天的人工智能助手，可以和人类进行对话。");

        if (!request.getQuestion().isEmpty()){
            roleContent = messages.get(messages.size() - 1);
            roleContent.setContent(request.getQuestion());
        }

        return SparkChatParam.builder()
                .messages(messages)
                .chatId(request.getUserId().toString())
                .build();
    }
}
